package com.iwhalecloud.uncc.mq.producer;

import com.ctg.mq.api.CTGMQFactory;
import com.ctg.mq.api.IMQProducer;
import com.ctg.mq.api.PropertyKeyConst;
import com.ctg.mq.api.bean.MQMessage;
import com.ctg.mq.api.bean.MQSendResult;
import com.ctg.mq.api.exception.MQProducerException;
import com.iwhalecloud.uncc.model.MqClusterConfig;
import com.iwhalecloud.uncc.utils.ZipUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.exception.ExceptionUtils;

import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
import java.util.UUID;

@Slf4j
public class CtgMqProducer implements MqProducer {

    private IMQProducer producer;

    @Override
    public void init(MqClusterConfig mqConfig) {
        try {
            if (log.isInfoEnabled()) {
                log.info("producer init...");
            }
            String producerGroup = UUID.randomUUID().toString();
            Properties p = new Properties();
            p.put(PropertyKeyConst.ProducerGroupName, producerGroup);
            p.put(PropertyKeyConst.NamesrvAddr, mqConfig.getNameserv());
            p.put(PropertyKeyConst.NamesrvAuthID, mqConfig.getUsername());
            p.put(PropertyKeyConst.NamesrvAuthPwd, mqConfig.getPwd());
            p.put(PropertyKeyConst.ClusterName, mqConfig.getClusterName());
            p.put(PropertyKeyConst.TenantID, mqConfig.getTenantId());
            p.put(PropertyKeyConst.SendMsgTimeout, mqConfig.getSendTimeout());
            producer = CTGMQFactory.createProducer(p);
            producer.connect();
            if (log.isInfoEnabled()) {
                log.info("producer init success!");
            }
        } catch (Exception e) {
            log.error("producer init exception!"+ ExceptionUtils.getStackTrace(e));
        }
    }

    @PreDestroy
    @Override
    public void destroy() {

    }

    /**
     * 发送普通消息
     *
     * @throws Exception
     */
    @Override
    public Object sendMsg(String topic, String tag, String key, String msg) throws Exception {
        MQMessage message = new MQMessage(topic, tag, key, msg.getBytes());
        MQSendResult result = producer.send(message);
        return result;
    }

    /**
     * 发送顺序消息
     */
    @Override
    public Object sendOrderlyMsg(String topic, String tag, String key,  String msg, String orderlyKey) throws Exception {
        MQMessage message = new MQMessage(topic, tag, key, msg.getBytes(), orderlyKey);
        MQSendResult result = producer.sendByGroupId(message);
        return result;
    }


    /**
     * 发送普通消息
     *
     * @throws Exception
     */
    @Override
    public Object sendMsg(String topic, String tag, String key, String msg, boolean compress) throws Exception {
        MQMessage message = new MQMessage(topic, tag, key, compress ? ZipUtil.zip(msg.getBytes(StandardCharsets.UTF_8)) : msg.getBytes(StandardCharsets.UTF_8));
        MQSendResult result = producer.send(message);
        return result;
    }

    @Override
    public Object sendMsg(String topic, String tag, String key, byte[] msgArr) throws Exception {
        MQMessage message = new MQMessage(topic, tag, key,msgArr);
        MQSendResult result = producer.send(message);
        return result;
    }

    /**
     * 发送顺序消息
     */
    @Override
    public Object sendOrderlyMsg(String topic, String tag, String key,  String msg, String orderlyKey, boolean compress) throws Exception {
        MQMessage message = new MQMessage(topic, tag, key, compress ? ZipUtil.zip(msg.getBytes(StandardCharsets.UTF_8)) : msg.getBytes(StandardCharsets.UTF_8), orderlyKey);
        MQSendResult result = producer.sendByGroupId(message);
        return result;
    }
}
